package com.atguigu.gmall.realtime.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.{DauInfo, PageLog}
import com.atguigu.gmall.realtime.utils.{BeanUtil, ESUtil, KafkaUtil, OffsetUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.lang
import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.util.Date
import scala.collection.mutable.ListBuffer

/**
 * @author caodan
 * @date 2022-09-05 17:25
 * @version 1.0
 *          日活宽表统计
 */
object DwdDauApp {


  def main(args: Array[String]): Unit = {
    //创建环境
    val conf = new SparkConf()
    conf.setMaster("local[1]").setAppName("dws_dau_app")
    val ssc = new StreamingContext(conf, Seconds(5))

    //主题
    val topicName: String = "DWD_PAGE_TOPIC"
    val groupId = "DWD_PAGE_TOPIC"

    // 从redis中获取偏移量
    val offsets: Map[TopicPartition, Long] = OffsetUtil.getOffset(topicName, groupId)

    var kafkaDstream: DStream[ConsumerRecord[String, String]] = null
    if (offsets != null && offsets.nonEmpty) {
      kafkaDstream = KafkaUtil.getKafkaDstream(topicName, ssc, groupId, offsets)
    } else {
      kafkaDstream = KafkaUtil.getKafkaDstream(topicName, ssc, groupId)
    }

    var offsetRanges: Array[OffsetRange] = null
    //获取偏移量
    kafkaDstream = kafkaDstream.transform(
      (rdd: RDD[ConsumerRecord[String, String]]) => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    // 类型转换
    val pageLogStream: DStream[PageLog] = kafkaDstream.map(
      (consumerRecord: ConsumerRecord[String, String]) => {
        val value: String = consumerRecord.value()
        val pageLog: PageLog = JSON.parseObject(value, classOf[PageLog])
        pageLog
      }
    )

    //获取数据 -- 自我审查 --> 三方审查
    val filterDStream: DStream[PageLog] = pageLogStream.filter(
      (pageLog: PageLog) => {
        pageLog.last_page_id == null
      }
    )
    // 三方审查
    val reviewAfterDstream: DStream[PageLog] = filterDStream.mapPartitions(
      (pageLogIter: Iterator[PageLog]) => {
        //获取jedis连接
        val jedis: Jedis = RedisUtil.getJedisConn
        val filterList: ListBuffer[PageLog] = ListBuffer[PageLog]()
        val pageLogList: List[PageLog] = pageLogIter.toList
        for (pageLog <- pageLogList) {
          val mid: String = pageLog.mid
          val sdf = new SimpleDateFormat("yyyy-MM-dd")
          val dateStr: String = sdf.format(new Date(pageLog.ts))
          val dauKey: String = s"DAU_USER_$dateStr"
          val long: lang.Long = jedis.sadd(dauKey, mid)
          if (long == 1L) {
            jedis.expire(dauKey, 24 * 3600)
            filterList.append(pageLog)
          }
        }
        RedisUtil.closeJedisConn(jedis)
        filterList.toIterator
      }
    )


    val dauInfos: DStream[DauInfo] = reviewAfterDstream.mapPartitions(
      (pageLogIter: Iterator[PageLog]) => {
        val jedis: Jedis = RedisUtil.getJedisConn
        val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        val dauInfoes: Iterator[DauInfo] = pageLogIter.map(
          (pageLog: PageLog) => {
            //1、将 pageLog 对象转换为DauInfo 对象
            val dauInfo = new DauInfo()
            BeanUtil.copyProperties(pageLog, dauInfo)

            // 2、从redis中获取用户维度数据
            val userid: String = pageLog.user_id
            // 拼接redis user key名
            val redisUserKey : String = s"DIM:user_info:$userid"
            val value: String = jedis.get(redisUserKey)
            val userJsonObjByUserId: JSONObject = JSON.parseObject(value)
            val gender: String = userJsonObjByUserId.getString("gender")
            val birthday: String = userJsonObjByUserId.getString("birthday")
            //当前时间 - birthday 时间为年龄
            val birthdayDate: LocalDate = LocalDate.parse(birthday)
            val nowDate: LocalDate = LocalDate.now()
            val period: Period = Period.between(birthdayDate, nowDate)
            val years: Int = period.getYears
            val age: String = years.toString
            dauInfo.user_gender = gender
            dauInfo.user_age = age

            // 从redis中获取身份维度数据
            val provinceId: String = pageLog.province_id
            val redisProKey : String = s"DIM:base_province:$provinceId"
            val redisProValue: String = jedis.get(redisProKey)
            val redisValueJsonByProId: JSONObject = JSON.parseObject(redisProValue)
            dauInfo.province_name = redisValueJsonByProId.getString("name")
            dauInfo.province_iso_code = redisValueJsonByProId.getString("iso_code")
            dauInfo.province_3166_2 = redisValueJsonByProId.getString("iso_3166_2")
            dauInfo.province_area_code = redisValueJsonByProId.getString("area_code")

            //日期处理
            val dateSdf: String = sdf.format(new Date(pageLog.ts))
            val dateArr: Array[String] = dateSdf.split(" ")
            val dateYear: String = dateArr(0)
            val hr: String = dateArr(1).split(":")(0)
            dauInfo.dt = dateYear
            dauInfo.hr = hr

            dauInfo
          }
        )
        RedisUtil.closeJedisConn(jedis)
        dauInfoes
      }
    )
    //写入es中
    dauInfos.foreachRDD(
      (rdd: RDD[DauInfo]) => {
        rdd.foreachPartition(
          (dauInfoter: Iterator[DauInfo]) => {
            val list: List[(String, DauInfo)] = dauInfoter.map((dauInfo: DauInfo) => (dauInfo.mid, dauInfo)).toList
            // 定义es中的index
            if(list != null && list.nonEmpty){
              val dauInfo: DauInfo = list.head._2
              val dt: String = dauInfo.dt
              val index : String = s"gmall_dau_info_$dt"
              ESUtil.bulkSaveIdempotent(list,index)
            }
          }
        )
        // 提交kafka offsets
        OffsetUtil.saveOffset(topicName,groupId,offsetRanges)
      }
    )

    //启动
    ssc.start()
    //阻塞
    ssc.awaitTermination()

  }

}
